package com.lry.rpc.netty.client;

import com.lry.common.CommonConstants;
import com.lry.event.ProviderNodeWrapper;
import com.lry.registy.AbstractRegister;
import com.lry.registy.RegistryFactory;
import com.lry.registy.URL;
import com.lry.rpc.config.NettyClientConfig;
import com.lry.rpc.config.RpcReferenceBean;
import com.lry.rpc.netty.BeanKey;
import com.lry.rpc.netty.Disposable;
import com.lry.util.CollectionUtils;
import com.lry.util.IpUtils;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.lry.common.CommonConstants.COLON;


@Slf4j
public class RpcClient implements Disposable, RemotingClient {

    protected ThreadPoolExecutor connectExecutor;

    private AbstractRegister registryService;

    private NettyClientConfig nettyClientConfig;

    private AtomicBoolean init = new AtomicBoolean(false);

    private RpcClientBootstrap bootstrap;

    public RpcClient(NettyClientConfig nettyClientConfig){
        this.nettyClientConfig = nettyClientConfig;
        this.registryService = RegistryFactory.getRegistry(nettyClientConfig.getRegitryType(),nettyClientConfig.getRegistryServiceAddress());
    }

    public RpcClient(){
        this(new NettyClientConfig());
    }

    public void subscribeService(RpcReferenceBean rpcReferenceBean){
        Class interfaceClass = rpcReferenceBean.getInterfaceClass();
        URL url = URL.builder().serviceName(interfaceClass.getName()).implName(rpcReferenceBean.getServiceImplName()).
                applicationName(nettyClientConfig.getApplicationName()).host(IpUtils.getIp()).build();

        url.addParam(CommonConstants.VERSION,rpcReferenceBean.getVersion());
        url.addParam(CommonConstants.GROUP,rpcReferenceBean.getGroup());
        url.addParam(CommonConstants.RPC_REFERENCE_BEAN,rpcReferenceBean);
        registryService.subscribe(url);
    }


    @Override
    public void start() {
        if(init.get()){
            log.warn("rpc client already starting....");
            return;
        }
        if(ClientCache.isReferenceBeanEmpty()){
            log.warn("rpc client start fail for subscribe list is empty");
            return;
        }
        init.compareAndSet(false,true);

        this.bootstrap = new RpcClientBootstrap(nettyClientConfig);

        this.connectExecutor = new ThreadPoolExecutor(5,
                10, 10, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>());

        connectServer();
    }

    public void connectServer(){
        for (BeanKey beanKey : ClientCache.beanKeySet()) {
            String serviceName = beanKey.getInterfaceName();
            String implName = beanKey.getImplName();
            CompletableFuture.runAsync(()->{
                List<String> ips = registryService.getProviderIps(serviceName,implName);
                while(CollectionUtils.isEmpty(ips)){
                    log.warn("通过serviceName={},implName={}获取不到ip，3s后重试",serviceName,implName);
                    try {
                        Thread.sleep(nettyClientConfig.getConnectGapMillis());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    ips = registryService.getProviderIps(serviceName,implName);
                }
                for (String ip : ips) {
                    String[] split = ip.split(COLON);
                    Channel channel = Connector.connect(InetSocketAddress.createUnresolved(split[0], Integer.parseInt(split[1])));

                    URL url = URL.builder().serviceName(serviceName).implName(implName).host(split[0]).port(Integer.parseInt(split[1])).build();
                    String providerContent = registryService.getProviderContent(url);
                    ProviderNodeWrapper providerNodeWrapper = URL.buildProviderNodeFromContent(providerContent);

                    ClientCache.putServiceNameAndChannel(implName,channel, providerNodeWrapper);
                }
                URL url = URL.builder().serviceName(serviceName).implName(implName).build();
                url.addParam(CommonConstants.IPS,ips);
                registryService.doAfterSubscribe(url);

            }, this.connectExecutor);
        }
    }

    @Override
    public void shutdown() {
        destroy();
    }

    @Override
    public void destroy() {

    }


}
